package com.xyw.code.bus.center.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.rabbitmq.client.Channel;
import com.xyw.code.bus.center.dao.SysBusMapper;
import com.xyw.code.bus.center.event.MQErrorReceiver;
import com.xyw.code.bus.center.event.MessageSendAsync;
import com.xyw.code.bus.center.service.ICommonMessageService;
import com.xyw.code.bus.common.entity.form.SendMessageForm;
import com.xyw.code.bus.common.entity.po.SysBus;
import com.xyw.code.bus.common.entity.po.SysBusLog;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * User: xuyiwei
 * Date: 2020/2/20
 * Time: 下午8:41
 * Email: 1328312923@qq.com
 * Description: No Description
 **/
@Service
@Slf4j
public class CommonMessageServiceImpl implements ICommonMessageService {

    @Autowired
    private SysBusMapper sysBusMapper;

    @Autowired
    private MessageSendAsync messageSendAsync;

    @Override
    public void dealFailMessage(Message message, Channel channel) {
        // redelivered = true, 表明该消息是重复处理消息
        Boolean redelivered = message.getMessageProperties().getRedelivered();
        /**
         * 这里对消息重入队列做设置，例如将消息序列化缓存至 Redis, 并记录重入队列次数
         * 如果该消息重入队列次数达到一次次数，比如3次，将不再重入队列，直接拒绝
         * 这时候需要对消息做补偿机制处理
         *
         * channel.basicNack与channel.basicReject要结合越来使用
         *
         */
        try {
            if (redelivered) {
                /**
                 * 1. 对于重复处理的队列消息做补偿机制处理
                 * 2. 从队列中移除该消息，防止队列阻塞
                 */
                // 消息已重复处理失败, 扔掉消息
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息
                log.error("消息[{}]重新处理失败，扔掉消息", new String(message.getBody()));
            }
            // redelivered != true,表明该消息是第一次消费
            if (!redelivered) {
                // 消息重新放回队列
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                log.error("消息[{}]处理失败，重新放回队列", new String(message.getBody()));
            }

        } catch (IOException e1) {
            e1.printStackTrace();
        }
    }

    @Override
    public void dealFailMessage(SendMessageForm sendMessageForm) {
        List<SysBus> sysBuss = sysBusMapper.selectList(Wrappers.<SysBus>lambdaQuery().eq(SysBus::getCorrelationId,sendMessageForm.getCorrelationId()));
        SysBus sysBus = null;
        if(sysBuss.size()>0){
            sysBus = sysBuss.get(0);
            //如果小于3 则继续发送
            sysBus.setErrorCount(sysBus.getErrorCount()+1);
            sysBus.updateById();
        }else{
            sysBus = new SysBus();
            sysBus.setMessage(sendMessageForm.getMessage());
            sysBus.setExchange(sendMessageForm.getExchange());
            sysBus.setRouteKey(sendMessageForm.getRouteKey());
            sysBus.setCorrelationId(sendMessageForm.getCorrelationId());
            sysBus.setStatus(-1);
            sysBus.setErrorCount(1);
            sysBus.insert();
        }

        if(sysBus.getErrorCount()<4){
            messageSendAsync.sendMsg(sendMessageForm.getExchange(),sendMessageForm.getRouteKey(),sendMessageForm.getMessage(),new CorrelationData(sendMessageForm.getCorrelationId()));
        }
        //更新错误日志
        SysBusLog sysBusLog = new SysBusLog();
        sysBusLog.setBusId(sysBus.getId());
        sysBusLog.setExceptionLog(sendMessageForm.getExceptionMsg());
        sysBusLog.insert();
     }
}
